package com.study.flink.batch;

import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;

/**
 * @author stephen
 * @create 2019-01-28 21:05
 * @since 1.0.0
 */
public class BatchDemoWithFirstN {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        ArrayList<Tuple2<Integer,String>> data1 = new ArrayList<>();
        data1.add(new Tuple2<>(1,"QQ"));
        data1.add(new Tuple2<>(2,"QQQ"));
        data1.add(new Tuple2<>(3,"QQQQ"));
        data1.add(new Tuple2<>(4,"QQQQQ"));
        data1.add(new Tuple2<>(5,"QQQQQQ"));
        data1.add(new Tuple2<>(6,"QQQQQQQ"));
        data1.add(new Tuple2<>(7,"QQQQQQQQ"));
        data1.add(new Tuple2<>(8,"QQQQQQQQQ"));
        data1.add(new Tuple2<>(9,"QQQQQQQQQQ"));

        DataSource<Tuple2<Integer,String>> source = env.fromCollection(data1);

        // 所有数据中的前2
        source.first(2).print();

        System.out.println("=======================");

        // 根据第一个字段排序，获取分组中前2两个元素
        source.groupBy(0).first(2).print();

        System.out.println("=======================");

        // 根据第一个字段排序，根据第二个字段排序，获取每组前2个元素
        source.groupBy(0).sortGroup(1, Order.DESCENDING).first(2).print();

        System.out.println("========================");

        // 不分组，在本地全局排序（数据量大可能存在问题），获取前2个
        source.sortPartition(1,Order.DESCENDING).first(2).print();

    }
}
